package com.atguigu.flink.wordcount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/2/23
 *
 *      flink可以进行批处理，替代hive。
 *      目前这部分api已经过时了(不推荐使用)！
 *               DataSet API
 *      以后希望使用进行批处理，官方推荐用 有界流！
 *    ------------------------
 *      ①创建程序运行的环境
 *              举例:  SparkContext
 *      ②使用环境获取数据，封装为框架可以计算的数据模型
 *              举例:  RDD
 *      ③对数据模型进行运输
 *            RDD = RDD.xxxx
 *
 */
public class Demo1_BatchDemo
{
    public static void main(String[] args) throws Exception {

        //创建程序运行的环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        //以计算的数据模型
        DataSource<String> dataSource = env.readTextFile("data/words.txt");

        //进行单词统计
        FlatMapOperator<String, Tuple2<String, Integer>> ds1 = dataSource
            .flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>()
            {
                /*
                    Collector<Tuple2<String, Integer>> out: 输出记录的收集器。
                            只需要把希望输出的数据，通过out进行收集，它会自动发送到下游处理！
                 */
                @Override
                public void flatMap(String line, Collector<Tuple2<String, Integer>> out) throws Exception {
                    String[] words = line.split(" ");
                    for (String word : words) {
                        out.collect(new Tuple2<>(word, 1));
                    }
                }
            });

        ds1
            //spark中 reduceByKey： 按照key分组，执行一个reduce函数
            // groupBy(int  pos): 针对Tuple类型进行分组，  groupBy(String)： 针对 POJO(Bean)类型分组
            .groupBy(0)
            .sum(1)
            .print();


    }
}
